package com.hngy.scala

import org.apache.spark.{SparkConf, SparkContext}

/**
  * reduce：聚合计算
  * collect：获取元素集合
  * take(n)：获取前n个元素
  * count：获取元素总数
  * saveAsTextFile：保存文件
  * countByKey：统计相同的key出现多少次
  * foreach：迭代遍历元素
  */
object ActionOpScala {

  def main(args: Array[String]): Unit = {
    //创建sparkContext
    val sc = getSparkContext

    //reduce：聚合计算
    //reduceOp(sc)
    //collect：获取元素集合
    //collectOp(sc)
    //take(n)：获取前n个元素
    //takeOp(sc)
    //count：获取元素总数
    //countOp(sc)
    //saveAsTextFile：保存文件
    //saveAsTextFileOp(sc)
    //countByKey：统计相同的key出现多少次
    //countByKeyOp(sc)
    //foreach：迭代遍历元素
    foreachOp(sc)

    //停止sparkContext
    sc.stop()
  }

  def foreachOp(sc: SparkContext): Unit = {
    val dataRDD = sc.parallelize(Array(1,2,3,4,5,6))
    dataRDD.foreach(println(_))
  }

  def countByKeyOp(sc: SparkContext): Unit = {
    val dataRDD = sc.parallelize(Array(("A",1001),("B",1002),("A",1003),("A",1004)))
    val res = dataRDD.countByKey()
    for ( (k,v)<-res){
      println(k + "," + v)
    }
  }

  def saveAsTextFileOp(sc: SparkContext): Unit = {
    val dataRDD = sc.parallelize(Array(1,2,3,4,5,6))
    //指定HDFS的路径信息即可，需要指定一个不存在的目录
    //dataRDD.saveAsTextFile("hdfs://hadoop001:9001/out2")
    //保存到本地
    dataRDD.saveAsTextFile("D:\\out2")
  }

  def countOp(sc: SparkContext): Unit = {
    val dataRDD = sc.parallelize(Array(1,2,3,4,5,6))
    val num = dataRDD.count()
    println(num)
  }

  def takeOp(sc: SparkContext): Unit = {
    val dataRDD = sc.parallelize(Array(1,2,3,4,5,6))
    var res = dataRDD.take(2)
    for (item <- res){
      println(item)
    }
  }

  def collectOp(sc: SparkContext): Unit = {
    val dataRDD = sc.parallelize(Array(1,2,3,4,5,6))
    //注意：如果RDD中数据量过大，不建议使用collect，因为最终的数据会返回给Driver进程
    //如果想要获取几条数据，查看一下数据格式，可以使用take(n)
    val res = dataRDD.collect()
    for ( item <- res ){
      println(item)
    }
  }

  def reduceOp(sc: SparkContext): Unit = {
    val dataRDD = sc.parallelize(Array(1,2,3,4,5,6))
    val num = dataRDD.reduce(_ + _)
    println(num)
  }

  private def getSparkContext: SparkContext = {
    //创建sparkContext
    val conf = new SparkConf()
    conf.setAppName("ActionOpScala")
      .setMaster("local")
    new SparkContext(conf)
  }
}
